/*
 * ===================================================================
 *  
 *      Filename: log_srv_plugin_file.c
 *
 *   Description: This is the main entry for the file modules.All of the 
 *                logs will be recorded in the log file.
 *
 *       Version:  1.0
 *        Create:  2010/07/13
 *
 *        Author:  Keven Zhu     kevenzhu@tencent.com
 *     Copyright:  Reserved by Tencent.com Co.ltd
 *   
 * ====================================================================
 */

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <netdb.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <signal.h>
#include <sys/types.h>
#include <linux/unistd.h>
#include <unistd.h>
#include "lib/list.h"
#include "lib/util.h"

#define  __USE_GNU
#include <sched.h>
#include <pthread.h>

/*****************************************************************/
//template for the plugin,we take the file hanler as an example
char plugin_file_name[]="file target";
/*will not be freed*/
pc_plugin_ops file_plugin_ops;

extern long send_counter;

pthread_t file_plugin_flushd;
/*define the local ring buffer*/
void ** local_buffer_list;
/*ring buffer start*/
int file_ringbuffer_start  = 0;
int file_ringbuffer_end    = 0;
int file_max_buffer_size   = 0;
int file_max_buffer_counter= 0;
int file_max_buffer_counter_mask = 0;

struct iovec *file_iov = NULL;

/*define the local mutex for the ring buffer*/
pthread_spinlock_t file_ringbuffer_spinlock;
/*define the local mutex for the local operation*/
pthread_mutex_t file_output_mutex;

int file_log_fd;
/*the handler was defined in the color fd,we just control the reopen operation*/
extern int color_log_fd;

int alarm_flush_queue = FALSE;

/*this function call only be called once*/
static int file_plugin_instance_running = 0;

int acquire_mutex_for_file_output()
{
    int iret = FALSE;
    if(!pthread_mutex_trylock(&file_output_mutex))
    {
        //success to get the lock
        iret = TRUE;
    }
    pthread_spin_unlock(&file_ringbuffer_spinlock);
    return iret;
} 

int release_output_mutex_and_send_buffer()
{
    int local_start =0,local_end = 0, iovec_counter = 0,i=0,icounter = 0,imultiloop = 0; 
    struct iovec* temp_iovec = file_iov;

    /*get the current begin and end index of the ringbuffer*/
    pthread_spin_lock(&file_ringbuffer_spinlock);
    /**/
    local_start = file_ringbuffer_start;
    local_end   = file_ringbuffer_end;
    pthread_spin_unlock(&file_ringbuffer_spinlock);
    i = local_end - local_start;

    /*1000 is assumed, but should less than IOV_MAX(1024)*/
    if(i<MAX_WORK_QUEUE&&(!alarm_flush_queue)) /*need to adjust the value*/
    {
        pthread_mutex_unlock(&file_output_mutex);
        return TRUE;
    }

    /*disable the alarm flag*/
    if(alarm_flush_queue)
        alarm_flush_queue = FALSE;
    
    while(i)
    {
        /*each time,we can xmit 1024 packets at most*/
        if(i>=1024)
        {
            imultiloop = 1024;
        }
        else
            imultiloop = i;

        icounter      = 0;
        iovec_counter = 0;
        temp_iovec    = file_iov;

        /*write these buffers to the file,send them in a burst*/
        while(icounter < imultiloop) 
        {
            temp_iovec->iov_base = (void*)((unsigned long)local_buffer_list[file_ringbuffer_start&file_max_buffer_counter_mask] + 3);//the first 2 bytes contain length
            #ifdef DEBUG_LOG
            printf("buff4 %p\n",temp_iovec->iov_base);
            #endif
            temp_iovec->iov_len  = *((short*)local_buffer_list[file_ringbuffer_start&file_max_buffer_counter_mask]) - 3;
            file_ringbuffer_start++;
            temp_iovec++;
            icounter++;
        }

        #ifdef DEBUG_LOG
        printf("file send %d packet\n",i);
        #endif
        /*make sure the packet counter is correct*/
        writev(file_log_fd,file_iov,i);

        pthread_mutex_lock(&counter_mutex);
        send_counter +=imultiloop;
        pthread_mutex_unlock(&counter_mutex);

        /*free the recv buffer*/
        temp_iovec = file_iov;
        for(;iovec_counter<imultiloop;iovec_counter++)
        {
            #ifdef DEBUG_LOG
            printf("buff5 %p\n",temp_iovec->iov_base-3);
            #endif
            free(((char*)temp_iovec->iov_base)-3);
            temp_iovec++;
        }
        i = i-imultiloop;
    }

    pthread_mutex_unlock(&file_output_mutex);

	return 0;
}
int  file_plugin_lock()
{
    //get the spin lock firstly
    if(!pthread_spin_lock(&file_ringbuffer_spinlock))
    {
        return TRUE;
    }
    else
    {
        return FALSE;
    }
}

int  file_plugin_unlock()
{
    if(!pthread_spin_unlock(&file_ringbuffer_spinlock))
    {
        return TRUE;
    }
    else
    {
        return FALSE;
    }
}

int  file_plugin_trylock()
{
    //get the spin lock firstly
    return pthread_spin_trylock(&file_ringbuffer_spinlock);
}
/*msg format
 * <length><color><seq><date><ipaddress><process id><severity><process name>:<msg body>
 */
enum plugin_ret file_process_log(char* msg)
{
    int filled_buffer = 0;
    //int msg_size      = *((int*)msg);

    #ifdef DEBUG_LOG
    printf("file process buff3 %p\n",msg);
    #endif
    //filter the packet to decide accept it or not
    //all the log msg are logging to file

    //copy the msg to the local buff, each plugin must have an local 
    //buffer,and store the buffer async!!!!!!!!  NOT BLOCK
    
    //get the spin lock firstly
    pthread_spin_lock(&file_ringbuffer_spinlock);

    //store the packet to the ring buffer
    filled_buffer = file_ringbuffer_end - file_ringbuffer_start;
    if(filled_buffer < file_max_buffer_counter)
    {
        local_buffer_list[file_ringbuffer_end&file_max_buffer_counter_mask] = msg;
        file_ringbuffer_end++;
    }
    else
    {
        /*hold this packet, and try to send the packet, busy wating*/
        while(TRUE)
        {
            if(acquire_mutex_for_file_output())
            {
                release_output_mutex_and_send_buffer();

                //get the spin lock firstly
                pthread_spin_lock(&file_ringbuffer_spinlock);

                //store the packet to the ring buffer
                filled_buffer = file_ringbuffer_end - file_ringbuffer_start;
                if(filled_buffer < file_max_buffer_counter)
                {
                    local_buffer_list[file_ringbuffer_end&file_max_buffer_counter_mask] = msg;
                    file_ringbuffer_end++;
                    pthread_spin_unlock(&file_ringbuffer_spinlock);
                    return PR_ACCEPT;
                }
                else
                {
                    pthread_spin_unlock(&file_ringbuffer_spinlock);
                }
            }
            pthread_spin_lock(&file_ringbuffer_spinlock);
        }
    }


    if(acquire_mutex_for_file_output())
    {
        release_output_mutex_and_send_buffer();
    }

    //all the msg are placed to the local
    //buffer firstly, then try to get the sem to output it to the 
    //target

	return PR_ACCEPT;
}

void *alarm_handler_flush_queue(void *argc)
{

    while(1)
    {
        usleep(3000000);
        //get the spin lock firstly
        pthread_spin_lock(&file_ringbuffer_spinlock);
        if(acquire_mutex_for_file_output())
        {
            alarm_flush_queue = TRUE;
            release_output_mutex_and_send_buffer();
        }
    }
}

/*return the plugin name to the main coreframe*/
char* file_plugin_name()
{
    return plugin_file_name;
}

int file_unload_plugin()
{
    printf("    file plugin try to free all the queued buffer\n");
    pthread_spin_lock(&file_ringbuffer_spinlock);
    while(file_ringbuffer_start != file_ringbuffer_end)
    {
        free((char*)local_buffer_list[file_ringbuffer_start&file_max_buffer_counter_mask]);
        file_ringbuffer_start++;
    } 
    pthread_spin_unlock(&file_ringbuffer_spinlock);
   
    printf("    file plugin cancel the flushd thread\n");
    pthread_cancel(file_plugin_flushd);

    printf("    file plugin close the file handler\n");
    /*close the file handler*/
    close(file_log_fd);

    printf("    file plugin free the file iovec\n");
    /*free the iovec*/
    free((char*)file_iov);

    /*free the memory*/
    /*for(ipos = 0;ipos<file_max_buffer_counter;ipos++)
    {
        free(local_buffer_list[ipos]);
    }*/

    printf("    file plugin try to free the buffer list point\n");
    /*free the list*/
    free((char*)local_buffer_list);

    printf("    file plugin destroy the lock\n");
    /*destroy the mutex and spinlock*/
    pthread_spin_destroy(&file_ringbuffer_spinlock);
    pthread_mutex_destroy(&file_output_mutex);
    file_plugin_instance_running = FALSE;
    printf("    success to unload the file plugin\n");
	return 0;
}

int reopen_local_file(int file_type,int name_size,char *file_name)
{
    int tmp_new_fd,tmp_delete_fd=-1;
    char *file_path = (char*)malloc(name_size);
    if(!file_path)
    {
        printf("failed to malloc the memory for the new path\n");
        return FALSE;
    }

    memcpy(file_path,file_name,name_size);

    if((tmp_new_fd=open(file_path,O_APPEND|O_CREAT|O_RDWR,S_IRWXU)) == -1)
    {
        perror("failed to open the file");
        return FALSE;
    }
    if(REOPEN_LOG_FILE == file_type)
    {
        /*replace it log file point*/
        tmp_delete_fd = file_log_fd;
        file_log_fd = tmp_new_fd;
    }
    else if(REOPEN_COLOR_FILE == file_type)
    {
        /*replace the color file point*/
        tmp_delete_fd = color_log_fd;
        color_log_fd = tmp_new_fd;
    }
    close(tmp_delete_fd);
    free(file_path);
    return OK;
}
/*the buffer_counter must be the power of 2*/
pc_plugin_ops *file_initialize_plugin(int buffer_size, int buffer_counter)
{
    if(file_plugin_instance_running)
        return NULL;
    file_plugin_instance_running = TRUE;
    local_buffer_list = (void**)calloc(buffer_counter,sizeof(void*));

    /*assign the funtion point*/
    file_plugin_ops.process_log   = file_process_log;
    file_plugin_ops.plugin_name   = file_plugin_name;
    file_plugin_ops.unload_plugin = file_unload_plugin;
    file_plugin_ops.lock          = file_plugin_lock;
    file_plugin_ops.unlock        = file_plugin_unlock;
    file_plugin_ops.trylock       = file_plugin_trylock;
    pthread_spin_init(&file_ringbuffer_spinlock,PTHREAD_PROCESS_SHARED);
    pthread_mutex_init(&file_output_mutex, NULL);

    /*it's better for us to do the self-safe power 2 check*/
    file_max_buffer_size         = buffer_size;
    file_max_buffer_counter      = buffer_counter;
    file_max_buffer_counter_mask = buffer_counter - 1;

    if((file_log_fd=open(DEFAULT_LOG_PATH,O_APPEND|O_CREAT|O_RDWR,S_IRWXU)) == -1)
    {
        perror("failed to open the file");
        return NULL;
    }

    /*check the return result*/
    file_iov = (struct iovec*)malloc(sizeof(struct iovec)*buffer_counter);
    
    if(pthread_create(&file_plugin_flushd,NULL,alarm_handler_flush_queue,NULL))
    {
        perror("failed to create the file flushd thread\n");
        return NULL;
    }
    
    return &file_plugin_ops;
}

